package com.aotain.project.apollo.boltv2;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * @ClassName: UDPFloodBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author zs
 * @date 2016年11月14日 
 *
 */
public class UDPFloodBolt extends AbstractBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 固定源60s内累计包数目标|源|端口
	 */
	private HashMap<String,Long> fixedSourceMap = new HashMap<String,Long>();

	/**
	 * 伪造源60s内累计包  目标|端口
	 */
	private HashMap<String,Long> forgedSourceMap = new HashMap<String,Long>();
	
	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> upStreamoctets = new HashMap<String,Long>();

	/**
	 * 攻击开始时间
	 */
	private HashMap<String,Long> attackStartMap = new HashMap<String,Long>();

	/**
	 * 攻击持续的上次更新时间
	 */
	private HashMap<String,Long> attackLasttimeMap = new HashMap<String,Long>();
	
	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;

	@Override
	public void execute(Tuple tuple) {		
		try {
			if(isTickTuple(tuple)) {
				Logger.getLogger(UDPFloodBolt.class).info(String.format("###DEBUG:UDPFLOOD START EMIT [%s]",tuple.getSourceStreamId()));
				emitCountingData(collector);
				Logger.getLogger(UDPFloodBolt.class).info(String.format("###DEBUG:UDPFLOOD END EMIT [%s]",tuple.getSourceStreamId()));
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {			
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("UDPFloodBolt throws an exception!",e);
		}
	}


	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{

			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idc = tuple.getStringByField("idcno");
			String type = tuple.getStringByField("type");
			String gis =  new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
			String dport = tuple.getStringByField("dport");
			String[] gisArray = gis.split("#");								
			String destAreaName = gisArray[0];
			String destGis = gisArray[1];
			String fixedMapKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc + "|isFixed";
			String forgedMapKey = dip + "|" + dport + "|" + idc + "|" + destAreaName + "|" + destGis;
			
		
			//如果是UDP ,类型为17， 按照上行包书累计计算
			if("17".equals(type)) {
				if(tuple.contains("upstreampacket") && tuple.contains("upstreamoctets")){				
					long uppacket = tuple.getLongByField("upstreampacket");	
					long upstreamoctets = tuple.getLongByField("upstreamoctets");
	
					if(fixedSourceMap.containsKey(fixedMapKey)) {
						fixedSourceMap.put(fixedMapKey,fixedSourceMap.get(fixedMapKey) + uppacket);
					}else{
						fixedSourceMap.put(fixedMapKey,uppacket);
					}
	
					if(forgedSourceMap.containsKey(forgedMapKey)) {					
						forgedSourceMap.put(forgedMapKey,forgedSourceMap.get(forgedMapKey) + uppacket);
					}else{
						forgedSourceMap.put(forgedMapKey,uppacket);
					}
					
					if(upStreamoctets.containsKey(fixedMapKey)) {
						upStreamoctets.put(fixedMapKey,upStreamoctets.get(fixedMapKey) + upstreamoctets);
					}else{
						upStreamoctets.put(fixedMapKey,upstreamoctets);
					}
					
					if(upStreamoctets.containsKey(forgedMapKey)) {
						upStreamoctets.put(forgedMapKey,upStreamoctets.get(forgedMapKey) + upstreamoctets);
					}else{
						upStreamoctets.put(forgedMapKey,upstreamoctets);
					}
	
				}
			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}


	public void emitCountingData(OutputCollector collector) {
		try {
			Logger.getRootLogger().error("fixedSourceMap.size()>>>"+fixedSourceMap.size());
			
			for(Entry<String, Long> entry : fixedSourceMap.entrySet()) {

				String entryKey = entry.getKey();
				long entryValue = entry.getValue();
				
				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];	
				if("183.62.140.140".equals(dip))		{
					Logger.getRootLogger().info("fixedSourceMap>>>entryValue >>>"+entryKey + ">>>" + entryValue);
				}
				
				long currenttime = System.currentTimeMillis();
				if(entryValue/60 >= 100000) {//如果每秒上行包总数大于10000则判断有攻击					
					
					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry,true);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,true,currenttime);
					}
				}else {
					//如果没大于门限 且过期3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {
							EndIntoData(entryKey,true);													
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
						}
					}
				}
			}
			
			Logger.getRootLogger().error("forgedSourceMap.size()>>>"+forgedSourceMap.size());

			//伪造源攻击判断
			for(Entry<String, Long> entry : forgedSourceMap.entrySet()) {

				String entryKey = entry.getKey();
				long entryValue = entry.getValue();
				
				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];
				
				long currenttime = System.currentTimeMillis();
				if(entryValue/60 >= 10000) {//如果每秒上行包总数大于10000则判断有攻击
					
					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry,false);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,false,currenttime);
					}
				}else {
					//如果没大于门限 也时间超过3分钟 则攻击结束
					
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {

							EndIntoData(entryKey,false);									
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
						}
					}
				}
			}

			ArrayList<String> removeItem = new ArrayList<String>();			
			for(Entry<String,Long> entry:attackLasttimeMap.entrySet()) {
				String entryKey = entry.getKey();
					
				if(System.currentTimeMillis() - entry.getValue() > END_DURATION) {					
					
					removeItem.add(entryKey);
					if(entryKey.endsWith("isFixed")) {	
						EndIntoData(entryKey,true);
					} else {
						EndIntoData(entryKey,false);
						
					}					
					attackStartMap.remove(entryKey);
				}
			}

			for(String key  : removeItem) {
				attackLasttimeMap.remove(key);
			}

			removeItem.clear();
			upStreamoctets.clear();
			fixedSourceMap.clear();
			forgedSourceMap.clear();
		} catch (Exception e ) {
			e.printStackTrace();
			Logger.getLogger(getClass()).error("UDPFlood Exception",e);;
		}
	}
	
	/**
	 * 新记录入库
	 * @Title: InsertIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author zs
	 * @date 2016年11月21日 
	 */
	//	public void InsertIntoData() {
	public void InsertIntoData(Entry<String,Long> entry,boolean isFixed,long time) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);

		String destAreaName = "";
		String destGis = "";
		String sourceAreaName = "";
		String sourceGis = "";
		String sourceAreaCountry = "";
		String sourceAreaId = "";		
		String sourceProvinceName =  "";; //如果省为空，精确到国家
		String sourceAreaCityId = "";
		String sourceAreaProvinceId = "";
		String dip = "";
		String sip = "";
		String dport = "";
		String gis = "";
		String idc = "";
		String desc = "";
		String rowKeyAbnormal = "";

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(time));
		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		String rowkey3 = String.format("%s_%s", accesstime, sip);
		
		String entryKey = entry.getKey();
		long entryValue = entry.getValue();
		
		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[2];
			gis = tuplesp[3];
			idc = tuplesp[4];
			String[] gisArray = gis.split("#");								
			destAreaName = gisArray[0];
			destGis = gisArray[1];
			sourceAreaName = gisArray[2];
			sourceGis = gisArray[3];
			sourceAreaCountry = gisArray[4];
			sourceAreaId = gisArray[5];		
			sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
			sourceAreaCityId = gisArray[7];
			sourceAreaProvinceId = gisArray[8];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_7";	
			desc = String.format("UDPFLOOD固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entryValue, upStreamoctets.get(entryKey)/(1024*60));
			
		} else {
			dip = tuplesp[0];
			sip = "伪造源";
			dport = tuplesp[1];
			idc = tuplesp[2];				
			destAreaName = tuplesp[3];;
			destGis = tuplesp[4];
			sourceAreaName = "伪造源";
			//				sourceGis = "ANY";
			sourceAreaCountry = "伪造源";
			sourceAreaId = "伪造源";	
			sourceProvinceName = "伪造源"; //如果省为空，精确到国家
			sourceAreaCityId = "伪造源";
			sourceAreaProvinceId = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport +  "_7";
			desc = String.format("UDPFLOOD伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]",entryValue, upStreamoctets.get(entryKey)/(1024*60));
		}

		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", dport);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", "7");
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(entryValue));
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(entryValue));

		hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
		hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(entryValue));
		hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
		//			    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", "");
		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "17");
		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);

		Date current = new Date();
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);

		collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				sourceProvinceName, 1l, dateStr_hour,"UDPFLOOD", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)entryValue));


	}

	/**
	 * 更新记录字段
	 * @Title: UpdateData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author zs
	 * @date 2016年11月21日 
	 */
	public void UpdateData(Entry<String,Long> entry,boolean isFixed) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);

		String dip = "";
		String sip = "";
		String desc = "";
		String dport = "";
		String rowKeyAbnormal = "";

		String entryKey = entry.getKey();
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		//		String rowkey3 = String.format("%s_%s", accesstime, sip);

		long entryValue = entry.getValue();
		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[2];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_7";	
			desc = String.format("UDPFLOOD固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entryValue, upStreamoctets.get(entryKey)/(1024*60));			
		} else {
			dip = tuplesp[0];
			dport = tuplesp[1];
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_7";
			desc = String.format("UDPFLOOD伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entryValue, upStreamoctets.get(entryKey)/(1024*60));
		}
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String idc = tuplesp[4];
		
	
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(entryValue));
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(entryValue));
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);

		Date current = new Date();
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
	    String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);
		
		collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				sourceProvinceName, 0l, dateStr_hour,"UDPFLOOD", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)entryValue));

		
	}

	/**
	 * 攻击结束记录入库
	 * @Title: EndIntoData 
	 * @Description: 攻击结束记录入库
	 * @param entryKey
	 * @return void    返回类型 
	 * @author zs
	 * @date 2016年11月21日 
	 */
	public void EndIntoData(String entryKey,boolean isFixed) {
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		String dip = "";
		String sip = "";
		String desc = "";
		String dport = "";
		String rowKeyAbnormal = "";

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		//		String rowkey3 = String.format("%s_%s", accesstime, sip);


		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[1];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_7";	
			desc = String.format("UDPFLOOD固定源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)-attackStartMap.get(entryKey))/1000); 
		} else {
			dip = tuplesp[0];
			dport = tuplesp[1];
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_7";
			desc = String.format("UDPFLOOD伪造源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)-attackStartMap.get(entryKey))/1000);
		}
		
		
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ENDTIME", new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackLasttimeMap.get(entryKey) + 180*1001)));
		
	}



	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour",
				"abntype","idcno","sourcecountry","streamoctets","streampacket"));

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60); //10秒统计一次
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub

	}

}
